/*
 * Copyright 1999-2018 Alibaba Group Holding Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.alibaba.nacos.client.naming.remote.gprc.redo;

import com.alibaba.nacos.api.PropertyKeyConst;
import com.alibaba.nacos.api.common.Constants;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.api.naming.pojo.ServiceInfo;
import com.alibaba.nacos.api.naming.utils.NamingUtils;
import com.alibaba.nacos.client.env.NacosClientProperties;
import com.alibaba.nacos.client.naming.remote.gprc.NamingGrpcClientProxy;
import com.alibaba.nacos.client.naming.remote.gprc.redo.data.BatchInstanceRedoData;
import com.alibaba.nacos.client.naming.remote.gprc.redo.data.InstanceRedoData;
import com.alibaba.nacos.client.naming.remote.gprc.redo.data.SubscriberRedoData;
import com.alibaba.nacos.client.utils.LogUtils;
import com.alibaba.nacos.common.executor.NameThreadFactory;
import com.alibaba.nacos.common.remote.client.Connection;
import com.alibaba.nacos.common.remote.client.ConnectionEventListener;
import com.alibaba.nacos.common.remote.client.RpcClient;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Naming client gprc redo service.
 *
 * <p>When connection reconnect to server, redo the register and subscribe.
 *
 * @author xiweng.yy
 */
public class NamingGrpcRedoService implements ConnectionEventListener {

    /**
     * redoService 线程名称
     */
    private static final String REDO_THREAD_NAME = "com.alibaba.nacos.client.naming.grpc.redo";

    /**
     * redoExecutor 线程池 corePoolSize 数量，默认为 1
     */
    private int redoThreadCount;

    /**
     * redoExecutor 调度时间，默认3秒
     */
    private long redoDelayTime;

    /**
     * 标记这个 instance 要注册，在连接断开之后通过 redoService 的定时任务重新注册
     * 标记这个 instance 要注销，可以在 redoService 定时任务重试
     * 定时任务在 {@link NamingGrpcRedoService#NamingGrpcRedoService(NamingGrpcClientProxy, NacosClientProperties)} 中添加的
     */
    private final ConcurrentMap<String, InstanceRedoData> registeredInstances = new ConcurrentHashMap<>();

    /**
     * 标记订阅缓存，用于client redo 定时任务 检查 订阅信息
     */
    private final ConcurrentMap<String, SubscriberRedoData> subscribes = new ConcurrentHashMap<>();

    private final ScheduledExecutorService redoExecutor;

    /**
     * 连接状态，通过建立连接 和 断开连接 里面的方法来进行维护的
     * 1.建立连接 {@link RpcClient#start()} 里面的连接事件  notifyConnected(take.connection); 方法 里面就有在维护这个值
     * 2.同理，断开连接的时候 也会维护该值
     */
    private volatile boolean connected = false;


    /**
     * NamingGrpcRedoService 构造函数
     *
     * @param clientProxy
     * @param properties
     */
    public NamingGrpcRedoService(NamingGrpcClientProxy clientProxy, NacosClientProperties properties) {
        // 根据配置属性里的值给参数赋初始值
        setProperties(properties);

        // 初始化 redoExecutor
        this.redoExecutor = new ScheduledThreadPoolExecutor(redoThreadCount, new NameThreadFactory(REDO_THREAD_NAME));

        /**
         * 定时调度 RedoScheduledTask   中添加线程，
         * 进入到 {@link RedoScheduledTask#run()} 方法中，完成具体的任务
         */
        this.redoExecutor.scheduleWithFixedDelay(new RedoScheduledTask(clientProxy, this), redoDelayTime, redoDelayTime,
                TimeUnit.MILLISECONDS);
    }

    /**
     * 根据配置属性里的值给参数赋初始值
     *
     * @param properties 属性配置
     */
    private void setProperties(NacosClientProperties properties) {

        /**
         * redoExecutor 调度时间，默认3秒
         */
        redoDelayTime = properties.getLong(PropertyKeyConst.REDO_DELAY_TIME, Constants.DEFAULT_REDO_DELAY_TIME);

        //  redoExecutor 线程池 corePoolSize 数量，默认为 1
        redoThreadCount = properties.getInteger(PropertyKeyConst.REDO_DELAY_THREAD_COUNT,
                Constants.DEFAULT_REDO_THREAD_COUNT);
    }

    public ConcurrentMap<String, InstanceRedoData> getRegisteredInstances() {
        return registeredInstances;
    }

    public boolean isConnected() {
        return connected;
    }

    @Override
    public void onConnected(Connection connection) {
        connected = true;
        LogUtils.NAMING_LOGGER.info("Grpc connection connect");
    }

    /**
     * grpc 长连接断开后的操作
     *
     * @param connection connection has disconnected
     */
    @Override
    public void onDisConnect(Connection connection) {
        connected = false;
        LogUtils.NAMING_LOGGER.warn("Grpc connection disconnect, mark to redo");

        // 设置当前 instance 在重试缓存map 的状态
        synchronized (registeredInstances) {
            registeredInstances.values().forEach(instanceRedoData -> instanceRedoData.setRegistered(false));
        }

        // 设置当前 instance 在订阅缓存map 的状态
        synchronized (subscribes) {
            subscribes.values().forEach(subscriberRedoData -> subscriberRedoData.setRegistered(false));
        }
        LogUtils.NAMING_LOGGER.warn("mark to redo completed");
    }

    /**
     * 标记这个 instance 要注册，在连接断开之后通过 redoService 的定时任务重新注册
     * <p>
     * Cache registered instance for redo.
     *
     * @param serviceName service name
     * @param groupName   group name
     * @param instance    registered instance
     */
    public void cacheInstanceForRedo(String serviceName, String groupName, Instance instance) {
        // 拼装出 kye
        String key = NamingUtils.getGroupedName(serviceName, groupName);

        // 封装 redo 对象
        InstanceRedoData redoData = InstanceRedoData.build(serviceName, groupName, instance);

        // 将 redo 对象添加到缓存中
        synchronized (registeredInstances) {
            registeredInstances.put(key, redoData);
        }
    }

    /**
     * Cache registered instance for redo.
     *
     * @param serviceName service name
     * @param groupName   group name
     * @param instances   batch registered instance
     */
    public void cacheInstanceForRedo(String serviceName, String groupName, List<Instance> instances) {
        String key = NamingUtils.getGroupedName(serviceName, groupName);
        BatchInstanceRedoData redoData = BatchInstanceRedoData.build(serviceName, groupName, instances);
        synchronized (registeredInstances) {
            registeredInstances.put(key, redoData);
        }
    }

    /**
     * 标记 instance 已经注册过
     * <p>
     * Instance register successfully, mark registered status as {@code true}.
     *
     * @param serviceName service name
     * @param groupName   group name
     */
    public void instanceRegistered(String serviceName, String groupName) {
        String key = NamingUtils.getGroupedName(serviceName, groupName);
        synchronized (registeredInstances) {
            InstanceRedoData redoData = registeredInstances.get(key);
            if (null != redoData) {
                // 标记 instance 已经注册过
                redoData.registered();
            }
        }
    }

    /**
     *  标记 instance 要注销，可以在 redoService 定时任务重试
     *
     * Instance deregister, mark unregistering status as {@code true}.
     *
     * @param serviceName service name
     * @param groupName   group name
     */
    public void instanceDeregister(String serviceName, String groupName) {
        String key = NamingUtils.getGroupedName(serviceName, groupName);
        synchronized (registeredInstances) {

            // 状态标记
            InstanceRedoData redoData = registeredInstances.get(key);
            if (null != redoData) {
                // 设置注销中
                redoData.setUnregistering(true);
                // 设置最终状态
                redoData.setExpectedRegistered(false);
            }
        }
    }

    /**
     *  标记 instance 已经注销
     *
     * Instance deregister finished, mark unregistered status.
     *
     * @param serviceName service name
     * @param groupName   group name
     */
    public void instanceDeregistered(String serviceName, String groupName) {
        String key = NamingUtils.getGroupedName(serviceName, groupName);
        synchronized (registeredInstances) {
            InstanceRedoData redoData = registeredInstances.get(key);
            if (null != redoData) {
                // 标记 instance 已经注销
                redoData.unregistered();
            }
        }
    }

    /**
     * Remove registered instance for redo.
     *
     * @param serviceName service name
     * @param groupName   group name
     */
    public void removeInstanceForRedo(String serviceName, String groupName) {
        String key = NamingUtils.getGroupedName(serviceName, groupName);
        synchronized (registeredInstances) {
            InstanceRedoData redoData = registeredInstances.get(key);
            if (null != redoData && ! redoData.isExpectedRegistered()) {
                registeredInstances.remove(key);
            }
        }
    }

    /**
     * Find all instance redo data which need do redo.
     *
     * @return set of {@code InstanceRedoData} need to do redo.
     */
    public Set<InstanceRedoData> findInstanceRedoData() {
        Set<InstanceRedoData> result = new HashSet<>();
        synchronized (registeredInstances) {
            for (InstanceRedoData each : registeredInstances.values()) {
                if (each.isNeedRedo()) {
                    result.add(each);
                }
            }
        }
        return result;
    }

    /**
     * 标记服务要订阅，在 redoService 的定时任务中重新订阅
     *
     * Cache subscriber for redo.
     *
     * @param serviceName service name
     * @param groupName   group name
     * @param cluster     cluster
     */
    public void cacheSubscriberForRedo(String serviceName, String groupName, String cluster) {
        String key = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), cluster);

        SubscriberRedoData redoData = SubscriberRedoData.build(serviceName, groupName, cluster);
        synchronized (subscribes) {
            // 标记订阅
            subscribes.put(key, redoData);
        }
    }

    /**
     * 标记服务已订阅
     *
     * Subscriber register successfully, mark registered status as {@code true}.
     *
     * @param serviceName service name
     * @param groupName   group name
     * @param cluster     cluster
     */
    public void subscriberRegistered(String serviceName, String groupName, String cluster) {
        String key = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), cluster);
        synchronized (subscribes) {
            SubscriberRedoData redoData = subscribes.get(key);
            if (null != redoData) {
                redoData.setRegistered(true);
            }
        }
    }

    /**
     * Subscriber deregister, mark unregistering status as {@code true}.
     *
     * @param serviceName service name
     * @param groupName   group name
     * @param cluster     cluster
     */
    public void subscriberDeregister(String serviceName, String groupName, String cluster) {
        String key = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), cluster);
        synchronized (subscribes) {
            SubscriberRedoData redoData = subscribes.get(key);
            if (null != redoData) {
                redoData.setUnregistering(true);
                redoData.setExpectedRegistered(false);
            }
        }
    }

    /**
     * Judge subscriber has registered to server.
     *
     * @param serviceName service name
     * @param groupName   group name
     * @param cluster     cluster
     * @return {@code true} if subscribed, otherwise {@code false}
     */
    public boolean isSubscriberRegistered(String serviceName, String groupName, String cluster) {
        String key = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), cluster);
        synchronized (subscribes) {
            SubscriberRedoData redoData = subscribes.get(key);
            return null != redoData && redoData.isRegistered();
        }
    }

    /**
     * Remove subscriber for redo.
     *
     * @param serviceName service name
     * @param groupName   group name
     * @param cluster     cluster
     */
    public void removeSubscriberForRedo(String serviceName, String groupName, String cluster) {
        String key = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), cluster);
        synchronized (subscribes) {
            SubscriberRedoData redoData = subscribes.get(key);
            if (null != redoData && ! redoData.isExpectedRegistered()) {
                subscribes.remove(key);
            }
        }
    }

    /**
     * Find all subscriber redo data which need do redo.
     *
     * @return set of {@code SubscriberRedoData} need to do redo.
     */
    public Set<SubscriberRedoData> findSubscriberRedoData() {
        Set<SubscriberRedoData> result = new HashSet<>();
        synchronized (subscribes) {
            for (SubscriberRedoData each : subscribes.values()) {
                if (each.isNeedRedo()) {
                    result.add(each);
                }
            }
        }
        return result;
    }

    /**
     * get Cache service.
     *
     * @return cache service
     */
    public InstanceRedoData getRegisteredInstancesByKey(String combinedServiceName) {
        return registeredInstances.get(combinedServiceName);
    }

    /**
     * Shutdown redo service.
     */
    public void shutdown() {
        LogUtils.NAMING_LOGGER.info("Shutdown grpc redo service executor " + redoExecutor);
        registeredInstances.clear();
        subscribes.clear();
        redoExecutor.shutdownNow();
    }

}
